package com.codejiwei.flink.sink;

import com.codejiwei.flink.entity.Vehicle;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author jiwei
 * @description
 * @date 2023/5/31 12:28
 */
public class VehicleJdbcSink extends RichSinkFunction<Vehicle> {
    private Connection conn;
    private PreparedStatement ps;
    @Override
    public void invoke(Vehicle value, Context context) throws Exception {
//        ps.setString(1, value.getVehicleId());
//        ps.setString(2, value.getVehicleType());
//        ps.setString(3, value.getVehicleBrand());
//        ps.setString(4, value.getLicensePlateNumber());
//        ps.execute();
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useSSL=false", "root", "root");
        ps = conn.prepareStatement("insert into vehicle values(?,?,?,?)");
    }

    @Override
    public void close() throws Exception {
        ps.close();
        conn.close();
    }
}
